package cn.doitedu.rtmk.demo7;


import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;


public class RuleModel_B_Calculator implements RuleCalculator {

    ValueState<Integer> cntState;
    Roaring64Bitmap preSelectUsersBitmap;
    JSONObject message;
    JSONObject paramJsonObject;

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson, Roaring64Bitmap preSelectUsersBitmap) throws Exception {

        // 参数解析
        paramJsonObject = JSON.parseObject(ruleParamJson);
        String ruleId = paramJsonObject.getString("rule_id");


        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-CNT", Integer.class));
        this.preSelectUsersBitmap = preSelectUsersBitmap;

        message = new JSONObject();
        message.put("rule_id", ruleId);

    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {
        long userId = userEvent.getUser_id();
        if (!preSelectUsersBitmap.contains(userId)) return;

        String eventId = userEvent.getEvent_id();

        // 从规则参数中，提取动态画像条件的参数
        JSONObject dynamicProfileCondition = paramJsonObject.getJSONObject("dynamic_profile_condition");
        String dynamicEventId = dynamicProfileCondition.getString("event_id");
        String dynamicOperator = dynamicProfileCondition.getString("operator");
        int compareValue = dynamicProfileCondition.getIntValue("compare_value");


        int oldValue = 0;
        if (dynamicEventId.equals(eventId)) {
            oldValue = cntState.value() == null ? 0 : cntState.value();
            cntState.update(oldValue + 1);
        }


        // 从规则参数中，提取触发事件条件的event_id
        String triggerEventId = paramJsonObject.getString("trigger_event_id");

        if (triggerEventId.equals(eventId)) {
            if (compare(cntState.value(), dynamicOperator,compareValue)) {
                message.put("user_id", userId);
                message.put("hit_time", userEvent.getAction_time());

                out.collect(message.toJSONString());
            }
        }


    }


    private boolean compare(int userValue,String operator,int conditionValue){

        boolean res = false;

        switch (operator) {
            case "=":
                res = userValue == conditionValue;
                break;
            case ">":
                res = userValue > conditionValue;
                break;

            case ">=":
                res = userValue >= conditionValue;
                break;
            case "<":
                res = userValue < conditionValue;
                break;

            case "<=":
                res = userValue <= conditionValue;
                break;

            case "!=":
                res = userValue != conditionValue;
                break;
        }

        return res;
    }


}
